package com.zhou.common.base.mq.mq;


import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.MessageQueue;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class CombineProducer {

    private OrderProducer orderProducer;

    private Producer producer;

    private DefaultMQProducer defaultProducer;


    private final String MESSAGE_PROPERTY_KEY = "processDetail";

    public CombineProducer(OrderProducer orderProducer, Producer producer, DefaultMQProducer defaultProducer) {
        this.orderProducer = orderProducer;
        this.producer = producer;
        this.defaultProducer = defaultProducer;
    }

    public void sendDelay(String body, String tag, String topic, long delay) throws Exception {
        if (producer != null) {
            Message message = new Message(topic, tag, body.getBytes(StandardCharsets.UTF_8));
            // 延迟消息
            if (delay > 0) {
                long deliveryTime = System.currentTimeMillis() + delay;
                message.setStartDeliverTime(deliveryTime);
            }
            try {
                SendResult result = producer.send(message);
                if (result == null) {
                    throw new ONSClientException("发送消息返回结果为空");
                }
                log.info("msg {} send succ, result: {}", message.getKey(), result);
            } catch (ONSClientException ex) {
                throw ex;
            }
        }
        if (defaultProducer != null) {
            try {
                org.apache.rocketmq.common.message.Message message1 = new org.apache.rocketmq.common.message.Message(topic, tag, body.getBytes(StandardCharsets.UTF_8));
                message1.setDelayTimeLevel(16);
                org.apache.rocketmq.client.producer.SendResult sendResult = defaultProducer.send(message1);
                log.info("msg {} send succ, result: {}", message1.getKeys(), sendResult);
            } catch (Exception e) {
                log.info("消息发送失败，", e);
                throw e;
            }
        }
    }

    public void sendOrder(String body, String tag, String topic, String shardingKey, long delay) throws Exception {
        if (orderProducer != null) {
            Message message = new Message(topic, tag, body.getBytes(StandardCharsets.UTF_8));
            //将流程单号作为Message的Key
            if (delay > 0) {
                long deliveryTime = System.currentTimeMillis() + delay;
                message.setStartDeliverTime(deliveryTime);
            }
            message.setKey(shardingKey);
            try {
                SendResult result = orderProducer.send(message, shardingKey);
                if (result == null) {
                    throw new ONSClientException("发送消息返回结果为空");
                }
                log.info("msg {} send succ, result: {}", message.getKey(), result);
            } catch (ONSClientException ex) {
                throw ex;
            }
        }
        if (defaultProducer != null) {
            try {
                org.apache.rocketmq.common.message.Message message1 = new org.apache.rocketmq.common.message.Message(topic, tag, body.getBytes(StandardCharsets.UTF_8));
                message1.setKeys(shardingKey);
                org.apache.rocketmq.client.producer.SendResult sendResult = defaultProducer.send(message1, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object o) {
                        Integer qID = (Integer) o;
                        MessageQueue queue = list.get(qID);
                        return queue;
                    }
                }, 1);
                log.info("msg {} send succ, result: {}", message1.getKeys(), sendResult);
            } catch (Exception e) {
                log.info("消息发送失败，", e);
                throw e;
            }
        }
    }


    public void send(String body, String tag, String shardingKey, String topic, String process) throws Exception {
        if (orderProducer != null) {
            Message message = new Message(topic, tag, body.getBytes(StandardCharsets.UTF_8));
            //将流程单号作为Message的Key
            message.setKey(shardingKey);
            //向Message中添加processDetail属性
            message.putUserProperties(MESSAGE_PROPERTY_KEY, process);
            try {
                SendResult result = orderProducer.send(message, shardingKey);
                if (result == null) {
                    throw new ONSClientException("发送消息返回结果为空");
                }
                log.info("msg {} send succ, result: {}", message.getKey(), result);
            } catch (ONSClientException ex) {
                throw ex;
            }
        }
        if (defaultProducer != null) {
            try {
                org.apache.rocketmq.common.message.Message message1 = new org.apache.rocketmq.common.message.Message(topic, tag, body.getBytes(StandardCharsets.UTF_8));
                message1.setKeys(shardingKey);
                message1.putUserProperty(MESSAGE_PROPERTY_KEY, process);
                org.apache.rocketmq.client.producer.SendResult sendResult = defaultProducer.send(message1, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object o) {
                        Integer qID = (Integer) o;
                        MessageQueue queue = list.get(qID);
                        return queue;
                    }
                }, 1);
                log.info("msg {} send succ, result: {}", message1.getKeys(), sendResult);
            } catch (Exception e) {
                log.info("消息发送失败，", e);
                throw e;
            }
        }
    }
}
